package io.openkunlun.javadsl.server;

import com.google.common.collect.Lists;
import com.google.protobuf.Empty;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import io.openkunlun.javadsl.DaprExceptionResolver;
import io.openkunlun.javadsl.v1.DaprAppGrpc;
import io.openkunlun.javadsl.v1.DaprAppProtos;
import io.openkunlun.javadsl.v1.DaprCommonProtos;

import java.util.*;
import java.util.stream.Collectors;

final class DaprApp extends DaprAppGrpc.DaprAppImplBase {

    private final String id;
    private final String host;
    private final int port;
    private final int ttl;
    private final int weight;
    private final int warmup;

    private final DaprExceptionResolver errorHandler;
    private final Map<String, BindingHandler> bindingHandlers;
    private final Map<String, InvocationHandler> invocationHandlers;

    DaprApp(String id,
            String host,
            int port,
            int ttl,
            int weight,
            int warmup,
            DaprExceptionResolver errorHandler,
            Map<String, BindingHandler> bindingHandlers,
            Map<String, InvocationHandler> invocationHandlers) {
        this.id = id;
        this.host = host;
        this.port = port;
        this.ttl = ttl;
        this.weight = weight;
        this.warmup = warmup;
        this.errorHandler = errorHandler;
        this.bindingHandlers = bindingHandlers;
        this.invocationHandlers = invocationHandlers;
    }

    String id() {
        return id;
    }

    String host() {
        return host;
    }

    int port() {
        return port;
    }

    int ttl() {
        return ttl;
    }

    int weight() {
        return weight;
    }

    int warmup() {
        return warmup;
    }

    /**
     * @return
     */
    Set<String> roles() {
        return invocationHandlers.values().stream().map(InvocationHandler::service).collect(Collectors.toSet());
    }

    /**
     * @return
     */
    Set<String> bindings() {
        return bindingHandlers.values().stream().map(BindingHandler::name).collect(Collectors.toSet());
    }

    @Override
    public void onInvoke(DaprCommonProtos.InvokeRequest request, StreamObserver<DaprCommonProtos.InvokeResponse> responseObserver) {
        InvocationHandler handler = invocationHandlers.get(request.getMethod());
        if (null == handler) {
            responseObserver.onError(new StatusRuntimeException(Status.NOT_FOUND));
            return;
        }

        try {
            DaprCommonProtos.InvokeResponse response = handler.handle(request);
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            responseObserver.onError(errorHandler.resolve(e));
        }
    }

    @Override
    public void onHealthCheck(DaprCommonProtos.HealthCheckRequest request, StreamObserver<DaprCommonProtos.HealthCheckResponse> responseObserver) {
        // todo
        try {
            responseObserver.onNext(DaprCommonProtos.HealthCheckResponse.newBuilder().build());
            responseObserver.onCompleted();
        } catch (Exception e) {
            responseObserver.onError(errorHandler.resolve(e));
        }
    }

    @Override
    public void listBinding(Empty request, StreamObserver<DaprAppProtos.ListBindingResponse> responseObserver) {
        try {

            DaprAppProtos.ListBindingResponse response = DaprAppProtos.ListBindingResponse.newBuilder()
                    .addAllBindings(bindingHandlers.keySet())
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            responseObserver.onError(errorHandler.resolve(e));
        }
    }

    @Override
    public void onBindingEvent(DaprAppProtos.BindingEventRequest request, StreamObserver<DaprAppProtos.BindingEventResponse> responseObserver) {

        BindingHandler handler = bindingHandlers.get(request.getName());
        if (null == handler) {
            // should not happen.
            responseObserver.onError(new BindingHandlerNotFoundException());
            return;
        }

        try {
            DaprAppProtos.BindingEventResponse response = handler.handle(request);
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            responseObserver.onError(errorHandler.resolve(e));
        }
    }

    static class Builder {

        private final String host;
        private final int port;

        private String id = UUID.randomUUID().toString();
        private int ttl;
        private int weight;
        private int warmup;

        private DaprExceptionResolver errorHandler;
        private Map<String, BindingHandler> bindingHandlers;
        private Map<String, InvocationHandler> invocationHandlers;

        Builder(String host, int port) {
            this.host = host;
            this.port = port;
        }

        public Builder setId(String id) {
            this.id = id;
            return this;
        }

        public Builder setTtl(int ttl) {
            this.ttl = ttl;
            return this;
        }

        public Builder setWeight(int weight) {
            this.weight = weight;
            return this;
        }

        public Builder setWarmup(int warmup) {
            this.warmup = warmup;
            return this;
        }

        /**
         * @param handler
         * @return
         */
        public Builder addHandler(DaprHandler handler) {
            if (handler instanceof InvocationHandler) {
                return addHandler((InvocationHandler) handler);
            } else if (handler instanceof BindingHandler) {
                return addHandler((BindingHandler) handler);
            } else {
                return this;
            }
        }

        /**
         * @param handlers
         * @return
         */
        public Builder addHandlers(DaprHandler... handlers) {
            Lists.newArrayList(handlers).forEach(this::addHandler);
            return this;
        }

        public Builder addHandler(BindingHandler handler) {
            if (null == bindingHandlers) {
                bindingHandlers = new HashMap<>();
            }

            bindingHandlers.put(handler.name(), handler);
            return this;
        }

        public Builder addHandlers(BindingHandler... handlers) {
            Lists.newArrayList(handlers).forEach(this::addHandler);
            return this;
        }

        public Builder addHandler(InvocationHandler handler) {
            if (null == invocationHandlers) {
                invocationHandlers = new HashMap<>();
            }

            for (String method : handler.methods()) {
                invocationHandlers.put(method, handler);
            }
            return this;
        }

        public Builder addHandlers(InvocationHandler... handlers) {
            Lists.newArrayList(handlers).forEach(this::addHandler);
            return this;
        }

        public Builder errorHandler(DaprExceptionResolver errorHandler) {
            this.errorHandler = errorHandler;
            return this;
        }

        public DaprApp build() {
            if (null == errorHandler) {
                errorHandler = DaprExceptionResolver.DEFAULT;
            }
            if (null == bindingHandlers) {
                bindingHandlers = Collections.emptyMap();
            }
            if (null == invocationHandlers) {
                invocationHandlers = Collections.emptyMap();
            }
            return new DaprApp(id, host, port, ttl, weight, warmup, errorHandler, bindingHandlers, invocationHandlers);
        }
    }
}
